package com.lunsheng.train

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * @Description
  * @author zhonglunsheng
  * @create 2020-02-11 0:50
  */
object TransformApp {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[3]").setAppName("WorldCountV2")
    val ssc = new StreamingContext(conf, Seconds(5))
    val lines = ssc.textFileStream("D:/temp")

    // 黑名单
    val blankList = List("ZS", "LS")

    val step = ssc.sparkContext.parallelize(blankList).map(x => (x, true))
    step.foreach(println)
    val result = lines.map(x => (x.split(",")(1), x)).transform(rdd => {
      rdd.leftOuterJoin(step).filter(x => {
        println(x)
        x._2._2.getOrElse(false) != true
      }).map(x => x._2._1)
    })

    result.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
